#pragma once
#include <muduo/protobuf/codec.h>
#include <muduo/protobuf/dispatcher.h>
#include <muduo/net/TcpServer.h>
#include <mqcomm/Helper.hpp>
#include <mqcomm/Log.hpp>
#include <mqcomm/Message.pb.h>
#include <mqcomm/protocol.pb.h>
#include <mqcomm/threadpool.hpp>
#include "consumer.hpp"
#include <iostream>
#include <vector>
#include <string>
#include <memory>
#include <mutex>
#include <condition_variable>
#include <unordered_map>
#include <functional>
using namespace std;

namespace rbmq {
    typedef shared_ptr<ProtobufCodec> ProtobufCodecPtr;
    typedef shared_ptr<mqNs::basicCommonResponse> basicCommonResponsePtr;
    typedef shared_ptr<mqNs::basicConsumerResponse> basicConsumerResponsePtr;
    class channel {
    public :
        typedef shared_ptr<channel> ptr;
        channel(const muduo::net::TcpConnectionPtr& conn, const ProtobufCodecPtr& codec) 
            :_connection(conn)
            ,_codec(codec)
            ,_channel_id(UuidHelper::uuid())
        
        {
        }

        // 获取信道id
        string getChannelId() {
            return _channel_id;
        }

        // 声明一个交换机
        bool declareExchange(const string& name, mqNs::ExchangeType type, bool durable, bool auto_delete, \
        google::protobuf::Map<string, string>& args) {
            // 构造一个声明交换机的请求对象，请求id使用uuid生成一个唯一的，信道id就是当前新到的id
            mqNs::declareExchangeRequest req;
            req.set_cid(_channel_id);
            string rid = UuidHelper::uuid();
            req.set_rid(rid);
            req.set_exchange_name(name);
            req.set_exchange_type(type);
            req.set_durable(durable);
            req.set_auto_delete(auto_delete);
            req.mutable_args()->swap(args);
            // 向服务器发送请求，使用_codec进行发送
            _codec->send(_connection, req); 
            // 等待服务器的响应，收到服务器的响应后再返回成功，否则返回失败
            // 使用waitResponse进行等待
            basicCommonResponsePtr resp = waitResponse(rid);
            if (resp->ok()) {
                LOG(Info) << "声明交换机成功, 交换机名称: " << req.exchange_name() << endl;
            }
            return resp->ok();
        }
        // 删除一个交换机
        void deleteExchange(const string& exchange_name) {
            // 和声明交换机的流程类似
            mqNs::deleteExchangeRequest req;
            req.set_cid(_channel_id);
            string rid = UuidHelper::uuid();
            req.set_rid(rid);
            req.set_exchange_name(exchange_name);
            // 发送请求
            _codec->send(_connection, req);
            waitResponse(rid);
        }

        // 声明一个队列
        bool declareQueue(const string& queue_name, bool durable, bool exclusive, bool auto_delete, \
        google::protobuf::Map<string, string>& args) {
            mqNs::declareQueueRequest req;
            req.set_cid(_channel_id);
            string rid = UuidHelper::uuid();
            req.set_rid(rid);
            req.set_queue_name(queue_name);
            req.set_durable(durable);
            req.set_exclusive(exclusive);
            req.set_auto_delete(auto_delete);
            req.mutable_args()->swap(args);
            // 发送请求
            _codec->send(_connection, req);
            // 等待服务器响应
            basicCommonResponsePtr resp = waitResponse(rid);
            if (resp->ok()) {
                LOG(Info) << "声明队列成功, 队列名称: " << queue_name << endl;
            }
            return resp->ok();
        }

        // 删除一个队列
        void deleteQueue(const string& queue_name) {
            mqNs::deleteQueueRequest req;
            req.set_cid(_channel_id);
            string rid = UuidHelper::uuid();
            req.set_rid(rid);
            req.set_queue_name(queue_name);
            // 发送请求
            _codec->send(_connection, req);
            waitResponse(rid);
        }

        // 队列的绑定
        bool queueBinding(const string& queue_name, const string& exchange_name, const string& binding_key) {
            mqNs::QueueBindingRequest req;
            req.set_cid(_channel_id);
            string rid = UuidHelper::uuid();
            req.set_rid(rid);
            req.set_queue_name(queue_name);
            req.set_exchange_name(exchange_name);
            req.set_binding_key(binding_key);
            _codec->send(_connection, req);
            basicCommonResponsePtr resp = waitResponse(rid);
            if (resp->ok()) {
                LOG(Info) << "绑定队列: " << queue_name << " 与交换机: " << exchange_name << "成功!" << endl;
            }
            return resp->ok();
        }

        // 队列的解除绑定
        bool queueUnBinding(const string& queue_name, const string& exchange_name) {
            mqNs::QueueUnBindingRequest req;
            req.set_cid(_channel_id);
            string rid = UuidHelper::uuid();
            req.set_rid(rid);
            req.set_queue_name(queue_name);
            req.set_exchange_name(exchange_name);
            _codec->send(_connection, req);
            basicCommonResponsePtr resp = waitResponse(rid);
            return resp->ok();            
        }

        // 消息的发布(消息是发布给交换机的，具体发布给哪一个队列是由交换机的绑定关系确定的)
        void basicPublish(const string& exchange_name, const mqNs::BasicProperties* bp, const string& body) {
            mqNs::basicPublshRequest req;
            req.set_cid(_channel_id);
            string rid = UuidHelper::uuid();
            req.set_rid(rid);
            req.set_exchange_name(exchange_name);
            req.set_body(body);
            if (bp) {
                req.mutable_properties()->set_id(bp->id());
                req.mutable_properties()->set_delivery_mode(bp->delivery_mode());
                req.mutable_properties()->set_routing_key(bp->routing_key());
            }
            // 发送请求
            _codec->send(_connection, req);
            waitResponse(rid);
        }

        // 消息的确认
        void basicAck(const string& msg_id) {
            // 如果收到了消息确认，那么肯定是当前的消费者收到的，所以也就不需要再传入队列名称
            if (_consumer.get() == nullptr) {
                LOG(Debug) << "消息确认失败, 没有对应的消费者!" << endl;
                return;
            }
            mqNs::basicAckRequest req;
            req.set_cid(_channel_id);
            string rid = UuidHelper::uuid();
            req.set_rid(rid);
            req.set_queue_name(_consumer->_qname);
            req.set_mesage_id(msg_id);
            // 发送请求
            _codec->send(_connection, req);
            waitResponse(rid);
        }

        // 消息订阅
        bool basicConsumer(const string& consumer_tag, const string& queue_name, bool auto_ack, const consumerCallback& callback) {
            // 一个信道只允许关联一个消费者，如果_consumer不为空，则不能在订阅
            if (_consumer.get() != nullptr) {
                LOG(Debug) << "不能重复相应, 当前消费者已订阅其他队列!" << endl;
                return false;
            }
            // 先构造请求，等待服务器响应
            mqNs::basicConsumerRequest req;
            req.set_cid(_channel_id);
            string rid = UuidHelper::uuid();
            req.set_rid(rid);
            req.set_consumer_tag(consumer_tag);
            req.set_queue_name(queue_name);
            req.set_auto_ack(auto_ack);
            _codec->send(_connection, req);
            // sleep(1000);
            // 等待服务器响应
            basicCommonResponsePtr resp = waitResponse(rid);

            if (resp->ok()) {
                // 收到服务器的成功响应后，需要构造一个消费者对象_赋值给_consumer
                _consumer = make_shared<rbmq::consumer>(consumer_tag, queue_name, auto_ack, callback);
                LOG(Info) << "消费者: " << consumer_tag << " 订阅队列: " << queue_name << "消息成功!" << endl;
                return true;
            }
            LOG(Debug) << "订阅消息失败, 服务器响应失败" << endl;
            return false;
        }

        // 取消订阅
        void basicConcel() {
            if (_consumer.get() == nullptr) {
                LOG(Debug) << "取消订阅失败, 没有找到消费者信息!" << endl;
                return;
            }
            // 直接使用成员变量_consumer里面的属性来构建请求
            mqNs::basicConcelRequest req;
            req.set_cid(_channel_id);
            string rid = UuidHelper::uuid();
            req.set_rid(rid);
            req.set_queue_name(_consumer->_qname);
            req.set_consumer_tag(_consumer->_tag);
            // 发送请求
            _codec->send(_connection, req);
            // 等待服务器响应
            basicCommonResponsePtr resp = waitResponse(rid);
            if (resp->ok()) {
                // 取消订阅后，还需要重置_consumer
                _consumer.reset();
            }
        }

        // 打开一个信道
        bool openChannel() {
            // 构造请求
            mqNs::openChannelRequest req;
            req.set_cid(_channel_id);
            string rid = UuidHelper::uuid();
            req.set_rid(rid);
            // 发送请求
            _codec->send(_connection, req);
            // 等待服务端相应
            basicCommonResponsePtr resp = waitResponse(rid);
            return resp->ok();
        }
        
        // 关闭一个信道
        void closeChannel() {
            // 构建请求
            mqNs::closeChannelRequest req;
            req.set_cid(_channel_id);
            string rid = UuidHelper::uuid();
            req.set_rid(rid);
            // 发送请求
            _codec->send(_connection, req);
            // 等待服务端响应
            waitResponse(rid);
        }

        ~channel() {
            // 如果当前的信道关联了消费者，还要取消这个消费者的订阅
            if (_consumer.get() != nullptr) {
                basicConcel();
            }
        }
    public :
        // 添加响应(向链接提供的接口，向哈希表中添加响应)
        void putbasicResponse(const basicCommonResponsePtr& resp) {
            // 先加锁
            unique_lock<mutex> lock(_mutex);
            // 向哈希表中插入对象
            _basic_resps.insert(make_pair(resp->rid(), resp));
            // 唤醒所有的阻塞
            _connd.notify_all();
        }

        // 链接收到消息推送后，需要通过信道找到对应的消费者对象，通过回调函数处理消息
        void consumer(const basicConsumerResponsePtr& resp) {
            // unique_lock<mutex> lock(_mutex);
            // 如果_consumer不为空，则调用其内部的回调函数处理消息
            if (_consumer.get() == nullptr) {
                LOG(Debug) << "处理消息失败, 没有找到对应的订阅者!" << endl;
                return ;
            }
            // 如果_consumer里面的消费者表示不等于resp里面的消费者标识，也不能处理
            if (_consumer->_tag != resp->consumer_tag()) {
                LOG(Debug) << "消息处理失败, 当前消费者标识与响应的消费者标识不一致!" << endl;
                return;
            }
            _consumer->_callback(resp->consumer_tag(), resp->mutable_properties(), resp->body());
        }
    private : 
        // 等待响应，等待指定请求的响应
        basicCommonResponsePtr waitResponse(const string& rid) {
            // 先加锁
            unique_lock<mutex> lock(_mutex);
            // 使用_conn进行条件等待，一直等待，直到在哈希表中找到rid对应的响应
            _connd.wait(lock, [&rid, this]() {
                return _basic_resps.find(rid) != _basic_resps.end();
            });
            // 从哈希表中返回对象，并从哈希表中删除该响应对象
            basicCommonResponsePtr resp = _basic_resps[rid];
            _basic_resps.erase(rid);
            return resp;
        }

    private :
        string _channel_id; // 信道id
        muduo::net::TcpConnectionPtr _connection; // 信道所关联的tcp网络连接
        ProtobufCodecPtr _codec; // protobuf协议处理对象
        consumer::ptr _consumer; // 信道所关联的消费者
        mutex _mutex; // 互斥锁
        condition_variable _connd; // 条件变量
        unordered_map<string, basicCommonResponsePtr> _basic_resps; // 请求id与响应的映射哈希表
    };

    // 信道管理类
    class channelManager {
    public :
        typedef shared_ptr<channelManager> ptr;
        channelManager() {

        }

        // 创建一个信道
        channel::ptr createChannel(const muduo::net::TcpConnectionPtr& conn, const ProtobufCodecPtr& codec) {
            // 先加锁
            unique_lock<mutex> lock(_mutex);
            // 构造对象
            channel::ptr chn = make_shared<channel>(conn, codec);
            // 插入进哈希表
            _channels.insert(make_pair(chn->getChannelId(), chn));
            return chn;
        }
        // 删除一个信道
        bool removeChannel(const string& channel_id) {
            unique_lock<mutex> lock(_mutex);
            auto it = _channels.find(channel_id);
            if (it == _channels.end()) {
                return false;
            }
            _channels.erase(channel_id);
            return true;
        }
        // 获取一个信道
        channel::ptr getChannel(const string& channel_id) {
            unique_lock<mutex> lock(_mutex);
            auto it = _channels.find(channel_id);
            if (it == _channels.end()) {
                LOG(Debug) << "获取信道的时候, 没有找到对应的信道!" << endl;
                return channel::ptr();
            }
            return it->second;
        }

    private :
        mutex _mutex;
        unordered_map<string, channel::ptr> _channels; // 信道id信道对象的映射哈希表
    };
};
